package cn.wangjie.spark.hbase

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{CellUtil, HBaseConfiguration}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 从HBase表读取数据：htb_wordcount
 */
object SparkReadHBase {
	
	def main(args: Array[String]): Unit = {
		
		// 构建SparkContext上下文实例对象
		val sc: SparkContext = {
			// 1.a 创建SparkConf对象，设置应用属性，比如应用名称和运行模式
			val sparkConf = new SparkConf()
				.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
				.setMaster("local[2]")
				// TODO: 设置使用Kryo 序列化方式
				.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
				// TODO: 注册序列化的数据类型
				.registerKryoClasses(Array(classOf[ImmutableBytesWritable], classOf[Result]))
			// 1.b 创建实例对象
			SparkContext.getOrCreate(sparkConf)
		}
		
		
		/*
			  def newAPIHadoopFile[K, V, F <: NewInputFormat[K, V]](
			      path: String,
			      fClass: Class[F],
			      kClass: Class[K],
			      vClass: Class[V],
			      conf: Configuration = hadoopConfiguration
			  ): RDD[(K, V)]
		 */
		
		// TODO: 思考 -> 从HBase表读取数据时，连接HBase依赖ZK，表的名称
		val conf: Configuration = HBaseConfiguration.create()
		// 设置HBase依赖ZK信息
		conf.set("hbase.zookeeper.quorum", "node1.itcast.cn")
		conf.set("hbase.zookeeper.property.clientPort", "2181")
		conf.set("zookeeper.znode.parent", "/hbase")
		// 设置表的名称
		conf.set(TableInputFormat.INPUT_TABLE, "htb_wordcount")
		
		val hbaseRDD: RDD[(ImmutableBytesWritable, Result)] = sc.newAPIHadoopRDD(
			conf, //
			classOf[TableInputFormat], //
			classOf[ImmutableBytesWritable], //
			classOf[Result]
		)
		println(s"Count = ${hbaseRDD.count()}")
		
		hbaseRDD.take(6).foreach{case(rowkey, result) =>
			println(s"RowKey = ${Bytes.toString(rowkey.get())}")
			result.rawCells().foreach{cell =>
				val cf = Bytes.toString(CellUtil.cloneFamily(cell))
				val column = Bytes.toString(CellUtil.cloneQualifier(cell))
				val value = Bytes.toString(CellUtil.cloneValue(cell))
				println(s"${cf}:${column} = ${value}")
			}
		}
		
		// 应用结束，关闭资源
		sc.stop()
	}
	
}
